package org.example.flinktest.apidemo;

import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.example.flinktest.bean.SensorReading;
import org.example.flinktest.operators.MakeSensorMapFunction;

/**
 * @author shihb
 * @date 2019/12/12 18:39
 */
public class SideOutputDemo {


  public static void main(String[] args) throws Exception {
    OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

    //环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    //设置时间触发
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

    //source

    DataStream<String> source = env.socketTextStream("10.0.75.2", 9999);

    //transform
    SingleOutputStreamOperator<SensorReading> dataStream = source
        .map(new MakeSensorMapFunction())
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(1)) {
              @Override
              public long extractTimestamp(SensorReading element) {
                return element.getTimestamp() * 1000;
              }
            })
        .process(new ZeroDegreeAlarm());

    DataStream<String> sideOutputStream = dataStream.getSideOutput(outputTag);
    //sink

    dataStream.print("input data");
    sideOutputStream.print("alarm data");

    env.execute("window job test");


  }

}

class ZeroDegreeAlarm extends ProcessFunction<SensorReading, SensorReading> {


  OutputTag<String> outputTag = new OutputTag<String>("side-output") {
  };

  @Override
  public void processElement(SensorReading value, Context ctx, Collector<SensorReading> out)
      throws Exception {
    if (value.getTemperature() > 32.0) {
      out.collect(value);
    } else {
      ctx.output(outputTag, value.getId() + "零度报警");
    }

  }
}
